import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @ClassName WindowWordCount
 * @Deacription TODO
 * @Author wushumin
 * @Date 2021-06-21 16:39
 * @Version 1.0
 **/
public class WindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = environment.socketTextStream("192.168.6.76", 9999)
                .flatMap(new SplitterFunc())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1);
//        streamOperator.print();
        streamOperator.writeAsText("sinkFile.txt");
        environment.execute("window wordCount");
    }

    public static class SplitterFunc implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] datas = value.split(" ");
            for (String data : datas) {
                out.collect(new Tuple2<>(data,1));
            }
        }
    }
}
